package gnss_plugs

import (
	gnssutils "dgo/goutils/gnss"
	"dgo/goutils/utils"
	"fmt"
	"net"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const (
	TYPE_UNKOWN      = 0
	TYPE_NTRIPSERVER = 1
	TYPE_NTRIPCLIENT = 2
)

var __ntripconn_handle_counter uint32 = 1

type GnssCasterConn struct {
	sendingflag       int32
	sendTaskStartFlag int32

	RecID         string // 数据库记录ID
	caster        *GnssCaster
	handle        uint32
	sessionid     string
	conn          net.Conn
	refcnt        int32
	shakeflag     byte // 0:未成功, 1:认证成功
	lock          *sync.RWMutex
	requestParser *gnssutils.NtripRequest
	sendTask      *utils.QueueTask

	reqDisFlag int32
	fromAddr   string
	disInfo    string
	idstr      string
	user       string
	pass       string

	state_string string

	sendQueue *utils.MemBlockSyncList
	closeFlag int32

	// 发送耗时
	lastsendconsume time.Duration
	recvbytes       int64
	sendbytes       int64
	connecttime     time.Time
	disconnecttime  time.Time
	lastsendtime    time.Time
	lastrecvtime    time.Time

	// 接收状态
	//recvstep int16
	//sendstep int16

	CheckRecvTimeOutFlag int // 检测超时未接收数据断开标记， 0不检查
	Tag                  int64

	Type   uint8 // 连接对象类型, casterNtripServer, CasterNtripClientMap
	TagPtr interface{}
}

func newCasterConn(_conn net.Conn, caster *GnssCaster) *GnssCasterConn {
	rval := &GnssCasterConn{
		handle:            atomic.AddUint32(&__ntripconn_handle_counter, 1),
		conn:              _conn,
		caster:            caster,
		reqDisFlag:        0,
		requestParser:     gnssutils.NewNtripRequest(),
		lock:              new(sync.RWMutex),
		sendTaskStartFlag: 0,
		sendTask:          nil,
	}
	rval.sendQueue = utils.NewMemBlockSyncListEx(caster.memPool)
	rval.sendQueue.ConfigMaxSize(caster.max_sending_size)
	rval.sessionid = fmt.Sprintf("%d_%s", rval.handle, utils.RandKeyString(12))
	return rval
}

func (this *GnssCasterConn) String() string {
	if this.Type == TYPE_NTRIPCLIENT {
		return fmt.Sprintf("GNSSCASTER-NTRIPCLT-%s-%s-%s-%s", this.sessionid, this.GetMountPoint(), this.GetNtripUser(), this.fromAddr)
	} else {
		return fmt.Sprintf("GNSSCASTER-NTRIPSVR-%s-%s-%s", this.sessionid, this.GetMountPoint(), this.fromAddr)
	}
}

func (this *GnssCasterConn) sendingBegin() bool {
	if this.closeFlag == 1 {
		return false
	}

	if atomic.LoadInt32(&this.sendingflag) == 1 {
		return false
	}

	if !atomic.CompareAndSwapInt32(&this.sendingflag, 0, 1) {
		return false
	}

	return true
}

func (this *GnssCasterConn) sendingEnd() {
	this.sendingflag = 0
}

func (this *GnssCasterConn) sendingOnce(endfn func(sender *GnssCasterConn)) bool {
	if utils.GoFunCatchException {
		defer utils.PanicHandler()
	}

	defer endfn(this)

	if this.sendQueue.Size() == 0 {
		return false
	}

	caster := this.caster

	//t1 := time.Now()
	for {
		if caster.isTerminated == 1 {
			break
		}
		if this.sendQueue.Size() == 0 {
			break
		}
		block := this.sendQueue.PopMemBlock()
		if block == nil {
			break
		}

		if this.closeFlag == 1 {
			break
		}

		//err := this.conn.Write(block.GetBytes())
		buf := block.GetBytes()
		_, err := this.DirectlySendBuffer(buf)
		if err != nil {
			break
		}
		caster.Monitor_Sending.DeltaBuf(buf)

		//if time.Since(t1).Milliseconds() > 200 { // 发送超过200ms退出  下次再发
		//	break
		//}
	}
	return true
}

// handleRecv最后执行
func (this *GnssCasterConn) doClose() {
	this.disconnecttime = time.Now()
	if this.caster.OnNtripConnDisconnected != nil {
		this.caster.OnNtripConnDisconnected(this)
	} else if this.caster.processor != nil {
		this.caster.processor.OnNtripConnDisconnected(this)
	}

	if this.shakeflag == 1 {
		this.shakeflag = 0
		if this.Type == TYPE_NTRIPCLIENT {
			atomic.AddInt32(&this.caster.NtripCltOnlineCnt, -1)
		} else if this.Type == TYPE_NTRIPSERVER {
			atomic.AddInt32(&this.caster.NtripSvrOnlineCnt, -1)
		}
	}

	if this.sendTask != nil {
		utils.GoFunc2(this.sendTask.Stop)
		this.sendTask = nil
		this.sendTaskStartFlag = 0
	}
}

func (this *GnssCasterConn) GetDisInfo() string {
	return this.disInfo
}

func (this *GnssCasterConn) SetDisInfo(reason string) {
	this.disInfo = reason
}

func (this *GnssCasterConn) SessionId() string {
	return this.sessionid
}

func (this *GnssCasterConn) GetMountPoint() string {
	return this.requestParser.MountPoint
}

func (this *GnssCasterConn) GetUserAgent() string {
	str := this.requestParser.Headers["user-agent"]
	str = utils.Trim(strings.TrimPrefix(str, "NTRIP"))
	return str
}

func (this *GnssCasterConn) GetUserXForwardUser() string {
	str := this.requestParser.Headers["user-agent"]
	str = utils.Trim(strings.TrimPrefix(str, "NTRIP"))
	return str
}

func (this *GnssCasterConn) GetNtripUser() string {
	if len(this.user) > 0 {
		return this.user
	}

	this.user, this.pass, _ = this.requestParser.DecodeNtripClientAuthInfo()
	return this.user
}

func (this *GnssCasterConn) GetSessionInfo() string {
	return fmt.Sprintf("online:%.2f(s), recv:%d b, send:%d b", time.Since(this.connecttime).Seconds(), this.recvbytes, this.sendbytes)
}

func (this *GnssCasterConn) RecvBytes() int64 {
	return this.recvbytes
}

func (this *GnssCasterConn) SendBytes() int64 {
	return this.sendbytes
}

func (this *GnssCasterConn) GetLastSentConsume() time.Duration {
	return this.lastsendconsume
}

func (this *GnssCasterConn) GetLastSentT() time.Time {
	return this.lastsendtime
}

func (this *GnssCasterConn) GetHandle() uint32 {
	return this.handle
}

func (this *GnssCasterConn) IsActive() bool {
	return this.reqDisFlag == 0
}

func (this *GnssCasterConn) FromAddr() string {
	return this.fromAddr
}

func (this *GnssCasterConn) GetRequestHeaderText() string {
	return this.requestParser.ToString()
}

func (this *GnssCasterConn) StatusString() string {
	if this.Type == TYPE_NTRIPCLIENT {
		return fmt.Sprintf("[NTRIPCLT][%s](%s), [%s]请求数据源:%s, 接收流量:%d b, 发送流量:%d b, statestr:%s",
			this.sessionid, this.FromAddr(), this.GetNtripUser(), this.GetMountPoint(), this.recvbytes, this.sendbytes, this.state_string)
	} else if this.Type == TYPE_NTRIPSERVER {
		return fmt.Sprintf("[NTRIPSERV][%s](%s),数据源:%s, 接收流量:%d b, 发送流量:%d b, statestr:%s",
			this.sessionid, this.FromAddr(), this.GetMountPoint(), this.recvbytes, this.sendbytes, this.state_string)
	} else {
		return fmt.Sprintf("[UNKOWN][%s](%s),接收流量:%d b, 发送流量:%d b, statestr:%s",
			this.sessionid, this.FromAddr(), this.recvbytes, this.sendbytes, this.state_string)
	}
}

func (this *GnssCasterConn) CheckRecvTimeOut(val time.Duration) bool {
	return time.Now().Sub(this.lastrecvtime) > val
}

func (this *GnssCasterConn) CheckSendTimeOut(val time.Duration) bool {
	return time.Now().Sub(this.lastsendtime) > val
}

func (this *GnssCasterConn) checkKickOut(isec int) {
	if this.Type == TYPE_NTRIPSERVER {
		tick := int(time.Now().Sub(this.lastrecvtime).Seconds())
		if tick > isec {
			this.RequestDisconnect(fmt.Sprintf("[Caster.NTRIPSVR](%s)未接收数据时间,超时:%d秒,请求断开, statestr:%s", this.FromAddr(), tick, this.state_string))
		}
	} else if this.Type == TYPE_NTRIPCLIENT {
		tick := int(time.Now().Sub(this.lastsendtime).Seconds())
		if tick > isec {
			str := fmt.Sprintf("[Caster.NTRIPCLT][%s](%s)超时(%d秒)未发送请求断开, 接收流量:%d b, 发送流量:%d b, statestr:%s",
				this.sessionid, this.FromAddr(), tick,
				this.recvbytes, this.sendbytes, this.state_string)
			this.RequestDisconnect(str)
		} else if this.CheckRecvTimeOutFlag == 1 {
			tick2 := int(time.Now().Sub(this.lastrecvtime).Seconds())
			if tick2 > isec {
				str := fmt.Sprintf("[Caster.NTRIPCLT][%s](%s)超时(%d秒)未接收数据，请求断开, 接收流量:%d b, 发送流量:%d b, statestr:%s",
					this.sessionid, this.FromAddr(), tick2,
					this.recvbytes, this.sendbytes, this.state_string)
				this.RequestDisconnect(str)
			}
		}
	} else {
		t := time.Since(this.lastrecvtime)
		tick := int(t.Seconds())
		if tick > isec {
			this.RequestDisconnect(fmt.Sprintf("[Caster.UNKOWN](%s)未接收数据时间,超时:%d秒,请求断开, statestr:%s", this.FromAddr(), tick, this.state_string))
		}
	}
}

func (this *GnssCasterConn) RequestDisconnect(disReason string) {
	if atomic.CompareAndSwapInt32(&this.reqDisFlag, 0, 1) {
		this.disInfo = disReason
	}
	conn := this.conn
	if conn != nil {
		conn.Close()
	}

}

func (this *GnssCasterConn) GetConnectTime() time.Time {
	return this.connecttime
}

func (this *GnssCasterConn) GetSendDelay() time.Duration {
	return time.Now().Sub(this.lastsendtime)
}

func (this *GnssCasterConn) GetSendDelaySecs() float64 {
	if this.lastsendtime.IsZero() {
		return 999
	}
	return time.Now().Sub(this.lastsendtime).Seconds()
}

func (this *GnssCasterConn) GetConnectSecs() int64 {
	if this.connecttime.IsZero() {
		return 0
	}

	if this.disconnecttime.IsZero() {
		return utils.SecondOf(this.connecttime, time.Now())
	} else {
		return utils.SecondOf(this.connecttime, this.disconnecttime)
	}

}

func (this *GnssCasterConn) onSendQueueWork(args ...interface{}) {
	buf := args[0].([]byte)
	pop_t := time.Now()
	n, err := this.DirectlySendBuffer(buf)
	if len(args) > 1 {
		if fn, ok := args[1].(func(n int, err error)); ok {
			fn(n, err)
		} else if fn, ok := args[1].(func(n int, pop_t time.Time, err error)); ok {
			fn(n, pop_t, err)
		}
	}
}

func (this *GnssCasterConn) PostSendBuffer2(buffer []byte, fn func(n int, err error)) {
	if this.sendTask == nil {
		if atomic.CompareAndSwapInt32(&this.sendTaskStartFlag, 0, 1) {
			tmp := utils.NewQueueTask(this.onSendQueueWork)
			tmp.ConfigIdleSleepTime(time.Millisecond)
			tmp.Start()
			this.sendTask = tmp
		} else {
			fmt.Fprintf(os.Stderr, "send err, send task is not ready!")
		}
	}

	this.sendTask.Push(buffer, fn)
}

func (this *GnssCasterConn) PostSendBuffer(buffer []byte) {
	this.PostSendBufferCb(buffer, nil)
}

func (this *GnssCasterConn) PostSendBufferCb(buffer []byte, onSent func(n int, pop_t time.Time, err error)) {
	if this.caster == nil {
		if onSent != nil {
			onSent(0, time.Now(), fmt.Errorf("连接已经断线"))
		}
		return
	}
	if this.sendTask == nil {
		if atomic.CompareAndSwapInt32(&this.sendTaskStartFlag, 0, 1) {
			tmp := utils.NewQueueTask(this.onSendQueueWork)
			tmp.ConfigIdleSleepTime(time.Millisecond)
			tmp.Start()
			this.sendTask = tmp
		} else {
			fmt.Fprintf(os.Stderr, "send err, send task is not ready!")
		}
	}
	if onSent != nil {
		this.sendTask.Push(buffer, onSent)
	} else {
		this.sendTask.Push(buffer)
	}

}

func (this *GnssCasterConn) SendBuffer(buf []byte) (int, error) {
	if this.caster == nil {
		return 0, fmt.Errorf("连接已经断线!")
	}
	if this.caster.SendQueueType == 1 {
		this.PostSendBuffer(buf)
		return len(buf), nil
	} else if this.caster.SendQueueType == 2 {
		n, err := this.sendQueue.Write(buf)
		if err != nil {
			this.RequestDisconnect(fmt.Sprintf("发送数据异常:%s", err.Error()))
		}
		return n, err
	} else {
		return this.DirectlySendBuffer(buf)
	}
}

func (this *GnssCasterConn) DirectlySendBuffer(buf []byte) (int, error) {
	if this.caster == nil {
		return 0, fmt.Errorf("连接已经断线!")
	}

	t1 := time.Now()
	n, err := this.conn.Write(buf)
	//this.sendstep = 3
	if err == nil {
		this.lastsendconsume = time.Since(t1)
		this.lastsendtime = time.Now()
		this.sendbytes += int64(n)
		this.caster.NetInfo.DoSendBytes(uint64(n))
	} else {
		this.caster.Monitor_SendFail.DeltaBuf(buf)
		this.RequestDisconnect(fmt.Sprintf("发送数据异常:%s", err.Error()))
	}
	return n, err
}

// 处理连接, 连接不再使用
func (this *GnssCasterConn) handleRecv() {
	defer func() {
		this.closeFlag = 1
		this.doClose()
		this.caster.removeOnline(this)
	}()
	this.reqDisFlag = 0
	this.lastsendtime = time.Now()
	this.lastrecvtime = time.Now()
	this.connecttime = time.Now()

	goid := utils.GetCurrentGoRoutineID()
	this.fromAddr = this.conn.RemoteAddr().String()

	var buffer []byte
	for {
		if this.caster.processor != nil && this.Type == TYPE_NTRIPCLIENT {
			buffer = this.caster.processor.MakeNtripCltRecvBuff()
		} else if this.caster.processor != nil && this.Type == TYPE_NTRIPSERVER {
			buffer = this.caster.processor.MakeNtripSvrRecvBuff()
		} else {
			buffer = make([]byte, 64) // 第一次连接接收少量数据, 避免非法连接
		}

		if buffer == nil {
			buffer = make([]byte, 1024)
		}
		//this.recvstep = 1
		this.state_string = fmt.Sprintf("[%s][Gid:%d]this.conn.Read(buffer)", utils.NowString(), goid)
		n, err := this.conn.Read(buffer)
		//this.recvstep = 2
		if err != nil {
			str := fmt.Sprintf("处理(%s)接收数据请求时出现了异常:%v ", this.conn.RemoteAddr().String(), err)
			this.RequestDisconnect(str)
			break
		}

		if this.caster.DebugMode == 1 {
			utils.Log.Debugf("[%s]recv:%s, hex:%s", this.fromAddr, string(buffer[:n]), utils.BufToHexStr(buffer[:n], 0, ""))
		}

		this.lastrecvtime = time.Now()
		this.recvbytes += int64(n)
		this.caster.NetInfo.DoRecvBytes(uint64(n))

		if this.Type == TYPE_NTRIPCLIENT {
			//this.recvstep = 3
			this.state_string = fmt.Sprintf("[%s][Gid:%d]OnNtripCltData", utils.NowString(), goid)
			r := gnssutils.BOOL_FALSE
			if this.caster.OnNtripCltData != nil {
				r = this.caster.OnNtripCltData(this, buffer, n, this.recvbytes, this.sendbytes)
			} else if this.caster.processor != nil {
				r = this.caster.processor.OnNtripCltData(this, buffer, n, this.recvbytes, this.sendbytes)
			}
			//this.recvstep = 4
			if r == gnssutils.BOOL_FALSE {
				this.RequestDisconnect("[数据异常-断线]:处理接收NtripClient NMEA数据非法" + utils.BufToHexStr(buffer[:n], 0, ""))
				break
			}
		} else if this.Type == TYPE_NTRIPSERVER {
			this.state_string = fmt.Sprintf("[%s][Gid:%d]OnNtripSvrData", utils.NowString(), goid)
			if this.caster.OnNtripSvrData != nil {
				this.caster.OnNtripSvrData(this, buffer, n)
			} else if this.caster.processor != nil {
				this.caster.processor.OnNtripSvrData(this, buffer, n)
			}
		} else {
			var r int
			var disStr string
			this.state_string = fmt.Sprintf("[%s][Gid:%d]FirstDecode", utils.NowString(), goid)
			nr := this.requestParser
			for _, data := range buffer[:n] {
				r = nr.InputBuffer(data)
				if r == gnssutils.DECODE_HEAD {
					if nr.DecodeHeader() == TYPE_UNKOWN {
						disStr = "[数据异常-断线](解码数据头未知):" + utils.BufToHexStr(buffer[:n], 0, "")
						r = gnssutils.DECODE_ERR
						break
					}

					this.Type = nr.GetRequestType()

					if this.Type == TYPE_NTRIPSERVER {
						//this.recvstep = 20
						this.state_string = fmt.Sprintf("[%s][Gid:%d]First-OnNtripSvrShake", utils.NowString(), goid)
						r := gnssutils.BOOL_FALSE
						if this.caster.OnNtripSvrShake != nil {
							r = this.caster.OnNtripSvrShake(this, nr)
						} else if this.caster.processor != nil {
							r = this.caster.processor.OnNtripSvrShake(this, nr)
						}
						if r == gnssutils.BOOL_FALSE {
							//this.recvstep = 21
							this.RequestDisconnect("[验证失败](NtripSvr):" + string(buffer[:n]))
							break
						}
						this.shakeflag = 1
						atomic.AddInt32(&this.caster.NtripSvrOnlineCnt, 1) // 认证成功
						//this.recvstep = 22
						continue
					}

					if this.Type == TYPE_NTRIPCLIENT {
						//this.recvstep = 30
						this.state_string = fmt.Sprintf("[%s][Gid:%d]First-OnNtripCltShake", utils.NowString(), goid)
						r := gnssutils.BOOL_FALSE
						if this.caster.OnNtripCltShake != nil {
							r = this.caster.OnNtripCltShake(this, nr)
						} else if this.caster.processor != nil {
							r = this.caster.processor.OnNtripCltShake(this, nr)
						}
						if r == gnssutils.BOOL_FALSE {
							//this.recvstep = 31
							this.RequestDisconnect("[验证失败](NtripClt):" + utils.BufToHexStr(buffer[:n], 0, ""))
							break
						}
						this.shakeflag = 1
						atomic.AddInt32(&this.caster.NtripCltOnlineCnt, 1) // 认证成功
						//this.recvstep = 32
					}
				} else if r == gnssutils.DECODE_NMEA {
					nmeastr := nr.GetBodyString()
					tmpBuf := []byte(nmeastr)
					//this.recvstep = 40
					this.state_string = fmt.Sprintf("[%s][Gid:%d]First-OnNtripCltData", utils.NowString(), goid)
					r := gnssutils.BOOL_FALSE
					if this.caster.OnNtripCltData != nil {
						r = this.caster.OnNtripCltData(this, tmpBuf, len(tmpBuf), this.recvbytes, this.sendbytes)
					} else if this.caster.processor != nil {
						r = this.caster.processor.OnNtripCltData(this, buffer, n, this.recvbytes, this.sendbytes)
					}

					if r == gnssutils.BOOL_FALSE {
						this.RequestDisconnect("[数据异常-断线]:处理接收NtripClient NMEA数据非法" + utils.BufToHexStr(buffer[:n], 0, ""))
						break
					}
					//this.recvstep = 41
				} else if r == gnssutils.DECODE_ERR {
					disStr = "解码异常:" + utils.BufToHexStr(buffer[:n], 0, "")
					break
				}
			} // end for

			if r == gnssutils.DECODE_ERR {
				this.RequestDisconnect(disStr)
				break
			}
		}

		//this.recvstep = 99
	}
}
